#include "ThreadPool.h"
#include "Server.h"
#include "TcpConnect.h"
#include "Log.h"

#include <time.h>
#include <sys/time.h>
#include <algorithm>

using namespace std;

ThreadPool::ThreadPool(const string &name):_name(name), _started(false){
	pthread_cond_init(&_cond, NULL);
	pthread_mutex_init(&_mutex, NULL);
}

ThreadPool::ThreadPool(): _started(false){
	_name = "noNameThreadPool";
	pthread_cond_init(&_cond, NULL);
	pthread_mutex_init(&_mutex, NULL);
}

ThreadPool::~ThreadPool(){
	pthread_cond_destroy(&_cond);
	pthread_mutex_destroy(&_mutex);
}

void ThreadPool::clean(){
	//delete all the Threads
}

int ThreadPool::funcInThread(){
	LOG_DEBUG_VA("ThreadPool::funcInThread()");
	return 0;
}

void ThreadPool::start(int number){
	assert(!_started);
	assert(_threads.empty());
	_started = true;
	_threadsNum = number;
	LOG_DEBUG_VA("threadPool %s start with number: %d", _name.c_str(), _threadsNum);
	char buf[16] = { 0 };
	for (int i = 0; i < _threadsNum; ++i) {
		snprintf(buf, 15, "_%d", i);
		Thread *t = new Thread(bind(&ThreadPool::funcInThread, this), name()+buf);
		_threads.push_back(t);
		t->start();
	}
	return;
}

void ThreadPool::stop(){
	_started = false;
	pthread_cond_broadcast(&_cond);
//	for_each(_threads.begin(), _threads.end(),[](Thread * t) {t->join();});
	for(auto iter = _threads.begin(); iter != _threads.end(); ++iter) {
		(*iter)->join();
	}
	return;
}

Thread *ThreadPool::getOneThread(){
	static int count = 0;
	count = count % _threadsNum;
	Thread *t = _threads[count];
	__sync_fetch_and_add(&count, 1);
	return t;
}


//TcpThreadPool

TcpThreadPool::TcpThreadPool(const string &name): ThreadPool(name){
}

int TcpThreadPool::threadsSerial = -1;

void TcpThreadPool::start(int number){
	assert(!started());
	assert(threadsEmpty());
	setStarted();
	setThreadsNum(number);
	LOG_DEBUG_VA("TcpThreadPool %s start with number: %d", name().c_str(), threadsNum());
	char buf[16] = { 0 };
	for (int i = 0; i < threadsNum(); ++i) {
		snprintf(buf, 15, "_%d", i);
		TcpThread *t = new TcpThread(bind(&TcpThreadPool::funcInThread, this), name()+buf);
		addThread(t);
		t->start();
	}
	return;
}

int TcpThreadPool::addTcpConnect(TcpConnect *tc){
	
	static int count = 0;
	count = count % threadsNum();

	LOG_DEBUG_VA("TcpThreadPool addTcpConnect in Thread %d, total Threads %d", count, threadsSerial);
	struct epoll_event ev;
	ev.data.u64 = 0UL;
	ev.data.fd = tc->getSocketFd();
	ev.events = EPOLLIN;

	int ret = Socket::tcpEpollAdd(_epfd[count], tc->getSocketFd(), &ev);
	if(ret < 0){
		LOG_DEBUG_VA("tcpEpollAdd error TcpThread threadsSerial %d epoll fd %d", count, _epfd[count]);
		return ret;
	}
	__sync_fetch_and_add(&count, 1);

	return ret;
}

int TcpThreadPool::funcInThread(){
	static int serial = 0;

//	int serial = __sync_add_and_fetch(&threadsSerial,1);
	serial = serial % threadsNum();

	_epfd[serial] = epoll_create(1024);

	if (-1 == _epfd[serial]) {
		LOG_ERROR_VA("epoll fd create error.");
		return -1;
	}
	LOG_DEBUG_VA("TcpThread threadsSerial %d epoll fd %d, total %d", serial, _epfd[serial], threadsSerial);
	int ret = 0;

	while (started()) {
		int numfd = Socket::tcpEpollWait(_epfd[serial], _epevents[serial], 1024, -1);
		if (numfd < 0) {
			LOG_ERROR_VA("epoll wait error TcpThread threadsSerial %d epoll fd %d, total %d", serial, _epfd[serial], threadsSerial);
			return numfd;
		}
		for (int i = 0; i < numfd; ++i) {
			if (_epevents[serial][i].events & EPOLLIN) {
				if(_epevents[serial][i].data.fd == _listenSock[serial]) {
					//just recv data, dont care new socket, i think there is not new socket
					LOG_ERROR_VA("TcpThread epoll add listen sock!.");
				}else{
					int fd = _epevents[serial][i].data.fd;
					char buf[16];
					int len = Socket::tcpRead(fd, buf, 10, 1000);
					if(len == 0){
						LOG_DEBUG_VA("read 0 bytes, close the fd: %d", fd);
						getServer()->resetConnect(fd);
						Socket::tcpClose(fd);
						continue;
					}
					LOG_DEBUG_VA("read %d bytes: %s", len, buf);
					//do something
					Socket::tcpWrite(fd, buf, 10, 1000);
					LOG_DEBUG_VA("write %d bytes: %s", len, buf);
				}
			}else if(_epevents[serial][i].events & EPOLLHUP) {
				LOG_ERROR_VA("fd: %d get EPOLLHUP.", _epevents[serial][i].data.fd);
				Socket::tcpClose(_epevents[serial][i].data.fd);
			}else if (_epevents[serial][i].events & EPOLLERR) {
				LOG_ERROR_VA("fd: %d get EPOLLERR.", _epevents[serial][i].data.fd);
				Socket::tcpClose(_epevents[serial][i].data.fd);
			}else if (_epevents[serial][i].events & EPOLLOUT) {
				LOG_ERROR_VA("fd: %d get EPOLLOUT.", _epevents[serial][i].data.fd);
				//doWriteEvent(_epevents[i].data.fd);
			}else {
				LOG_ERROR_VA("fd: %d get something wrong.", _epevents[serial][i].data.fd);
				Socket::tcpClose(_epevents[serial][i].data.fd);
			}
		}
	}

	LOG_DEBUG_VA("TcpThreadPool::funcInThread()");
	return 0;
}

